package com.remoter.transport.socket.client;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import com.google.common.util.concurrent.SettableFuture;
import com.remoter.basic.bean.support.RemoterRequest;
import com.remoter.basic.bean.support.RemoterResponse;
import com.remoter.basic.transport.support.AbstractClient;
import com.remoter.basic.util.ServiceLoader.MetaName;
import com.remoter.transport.socket.client.handler.ClientMessageHandler;
import com.remoter.transport.socket.codec.MessageDecoder;
import com.remoter.transport.socket.codec.MessageEncoder;
import com.remoter.transport.socket.util.FinalTransportSocket;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * @author			koko
 * @date 			2017-08-10 13:37:25
 * @description 	类功能说明
 */
@MetaName(FinalTransportSocket.D_VAL_SPI_NAME)
public class SocketClient extends AbstractClient{
	
	protected InetSocketAddress remote;
	protected InetSocketAddress local;
	
	protected int connectTimeout;
	protected int readTimeout;
	protected int writeTimeout;
	
	protected NioEventLoopGroup work;
	protected ClientMessageHandler messageHandler;
	
	public SocketClient() {
		super();
		this.connectTimeout = configuration.getOption(FinalTransportSocket.O_VAL_CLIENT_CONNECT_TIMEOUT);
		this.readTimeout = configuration.getOption(FinalTransportSocket.O_VAL_CLIENT_READ_TIMEOUT);
		this.writeTimeout = configuration.getOption(FinalTransportSocket.O_VAL_CLIENT_WRITE_TIMEOUT);
	}

	@Override
	public void connect(ExecutorService executorService,final InetSocketAddress remote) throws Exception {
		if(null == remote){
			throw new IllegalArgumentException("remote address is null");
		}
		this.work = new NioEventLoopGroup(0,executorService);
		this.remote = remote;
		Bootstrap bootstrap = new Bootstrap();
		bootstrap.group(this.work);
		bootstrap.channel(NioSocketChannel.class);
		bootstrap.option(ChannelOption.SO_KEEPALIVE,true);
		bootstrap.option(ChannelOption.ALLOCATOR,PooledByteBufAllocator.DEFAULT);
		bootstrap.handler(new ChannelInitializer<SocketChannel>(){
			@Override
			protected void initChannel(SocketChannel socketChannel) throws Exception {
				ChannelPipeline channelPipeline = socketChannel.pipeline();
				channelPipeline.addLast(new MessageDecoder(serialization,RemoterResponse.class));
				channelPipeline.addLast(new MessageEncoder(serialization,RemoterRequest.class));
				channelPipeline.addLast(new ClientMessageHandler());
			}
		});
		logger.info("begin connect :" + this.remote);
		ChannelFuture channelFuture = bootstrap.connect(this.remote);
		boolean result = channelFuture.awaitUninterruptibly(this.connectTimeout,TimeUnit.MILLISECONDS);
		if(result && channelFuture.isSuccess()){
			local = (InetSocketAddress)channelFuture.channel().localAddress();
			messageHandler = channelFuture.channel().pipeline().get(ClientMessageHandler.class);
			logger.info("start success ["+ local.toString() +"]!");
		}else{
			logger.debug("consumer offline , disconnect " + remote);
			disConnect();
		}
	}
	
	@Override
	public void disConnect() throws Exception {
		if(null != this.work){
			this.work.shutdownGracefully().awaitUninterruptibly();
		}
		if(null != this.messageHandler){
			this.messageHandler.close();
		}
	}

	@Override
	public boolean isAvailable() {
		if(null == this.messageHandler){
			return false;
		}
		return this.messageHandler.isAvailable();
	}

	@Override
	public InetSocketAddress getRemoteAddress() {
		return this.remote;
	}

	@Override
	public InetSocketAddress getLocalAddress() {
		return this.local;
	}

	@Override
	public RemoterResponse sendMessage(RemoterRequest remoterRequest) throws Exception {
		if(!this.isAvailable()){
			RemoterResponse remoterResponse = new RemoterResponse(remoterRequest.getToken());
			remoterResponse.setException("client not available");
			return remoterResponse;
		}
		try{
			remoterRequest.setFromHost(this.local.getHostString());
			remoterRequest.setFromPort(this.local.getPort());
			remoterRequest.setToHost(this.remote.getHostString());
			remoterRequest.setToPort(this.remote.getPort());
			remoterRequest.setProtocol(FinalTransportSocket.D_VAL_SPI_NAME);
			SettableFuture<RemoterResponse> future = this.messageHandler.sendRemoterRequest(this.writeTimeout,remoterRequest);
			return future.get(this.readTimeout,TimeUnit.MILLISECONDS);
		}catch(Exception e){
			RemoterResponse remoterResponse = new RemoterResponse(remoterRequest.getToken());
			remoterResponse.setException("connect timeout");
			return remoterResponse;
		}
	}
	
}